/**
* Copyright © 2018 深圳市巨鼎医疗设备有限公司
*/
package com.zl.websocket.service.impl;

import java.util.Iterator;
import java.util.Map;
import java.util.Set;

import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Service;

import com.google.common.collect.Maps;
import com.google.common.collect.Sets;
import com.zl.websocket.constants.WebsocketConstants;
import com.zl.websocket.entity.Message;
import com.zl.websocket.entity.WebsocketMessage;
import com.zl.websocket.entity.WebsocketMessageTransform;
import com.zl.websocket.service.MessageService;
import com.zl.websocket.service.RouteService;
import com.zl.websocket.service.WebsocketSender;

/**
 * 使用websocket给前端发消息实现
 * 消息直接发布（publish）到redis特定的channel上，某个应用实例订阅（subscribe）这个channel接收到这个消息，然后调用websocketTemlate发送给前端
 * 
 * @author Zhouych
 * @Date: 2018年6月27日 下午1:55:54
 * @since JDK 1.8
 */
@Service("websocketMessageService")
public class WebsocketMessageServiceImpl implements MessageService {

	private static final Logger logger = LoggerFactory.getLogger(WebsocketMessageServiceImpl.class);

	@Autowired
	@Qualifier(value = "redisTemplate")
	private RedisTemplate<String, Object> redisTemplate;

	@Autowired
	private RedisTemplate<String, String> stringRedisTemplate;

	@Autowired
	@Qualifier("websocketRouteService")
	private RouteService websocketRouteService;

	@Autowired
	private WebsocketSender websocketSender;

	@Override
	public void send(Message message) {
		WebsocketMessage payload;
		if (message instanceof WebsocketMessage) {
			payload = (WebsocketMessage) message;
		} else {
			logger.error("请发送WebsocketMessage类型的消息");
			throw new RuntimeException("请发送WebsocketMessage类型的消息");
		}

		WebsocketMessageTransform msg = new WebsocketMessageTransform(payload);
		// 拿到所有的应用实例订阅地址
		Set<String> channels = stringRedisTemplate.opsForSet().members(WebsocketConstants.WEBSOCKET_CHANNELS);
		if (null != channels && !channels.isEmpty()) {
			for (String channel : channels) {
				doSend(channel, msg);
			}
		}
	}

	@Override
	public void send(String userId, Message message) {
		if (StringUtils.isBlank(userId)) {
			logger.error("用户ID不合法");
			throw new RuntimeException("用户ID不合法");
		}
		send(Sets.newHashSet(userId), message);
	}

	@Override
	public void send(Set<String> userIds, Message message) {
		WebsocketMessage payload;
		if (message instanceof WebsocketMessageTransform) {
			payload = (WebsocketMessage) message;
		} else {
			logger.error("请发送WebsocketMessage类型的消息");
			throw new RuntimeException("请发送WebsocketMessage类型的消息");
		}

		Map<String, Set<String>> channelUserMap = Maps.newHashMap();
		if (null != userIds && !userIds.isEmpty()) {
			for (String userId : userIds) {
				if (StringUtils.isBlank(userId)) {
					continue;
				}
				String channel = (String) stringRedisTemplate.opsForHash()
						.get(WebsocketConstants.WEBSOCKET_USER_CHANNEL, userId);
				if (StringUtils.isNotBlank(channel)) {
					if (channelUserMap.containsKey(channel)) {
						channelUserMap.get(channel).add(userId);
					} else {
						channelUserMap.put(channel, Sets.newHashSet(userId));
					}
				}
			}
		}
		for (Iterator<String> it = channelUserMap.keySet().iterator(); it.hasNext();) {
			String channel = it.next();
			Set<String> uids = channelUserMap.get(channel);
			WebsocketMessageTransform msg = new WebsocketMessageTransform(uids, payload);
			doSend(channel, msg);
		}
	}

	/**
	 * 如果是本机实例，直接调用websocket通道发送消息，否则发布到对应的redis channel，让订阅了该channel的服务实例负责发送消息
	 * 
	 * @param channel
	 * @param msg
	 */
	private void doSend(String channel, WebsocketMessageTransform msg) {
		if (websocketRouteService.getInstanceId().equals(channel)) {
			websocketSender.send(msg);
		} else {
			redisTemplate.convertAndSend(channel, msg);
		}
	}

	@Override
	public void onUserConnect(String id) {
		// TODO 建立连接成功的业务逻辑，比如发送失败的消息重发
	}

	@Override
	public void onUserDisConnect(String id) {
		// TODO 用户断开连接的业务逻辑，比如记录用户登出时间

	}

	@Override
	public void onSendFail(String userId, Message message) {
		// TODO 消息发送失败的业务逻辑，比如重发或者保存消息待用户登录再发

	}

}
